package com.yuki312.backoff.rxjava2;

import com.yuki312.backoff.core.Backoff;
import com.yuki312.backoff.core.BackoffAlgorithm;
import com.yuki312.backoff.core.ExponentialAlgorithm;
import com.yuki312.backoff.core.FixedIntervalAlgorithm;
import com.yuki312.backoff.core.RandomIntervalAlgorithm;

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Scheduler;
import io.reactivex.rxjava3.functions.BiConsumer;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.functions.Predicate;
import io.reactivex.rxjava3.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;


import org.reactivestreams.Publisher;

public class RxBackoff {

    private final Backoff backoff;
    private final Scheduler intervalScheduler;
    private static BackoffAlgorithm selectedAlgorithm;
    private static int mxRetryCount;

    private Predicate<Throwable> filter = new Predicate<Throwable>() {
        @Override
        public boolean test(Throwable throwable) throws Exception {
            return true;
        }
    };

    private BiConsumer<Throwable, Integer> onRetry = new BiConsumer<Throwable, Integer>() {
        @Override
        public void accept(Throwable e, Integer retry) throws Exception {
            // no-op
        }
    };

    private Consumer<Throwable> onAbort = new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            // no-op
        }
    };

    public static RxBackoff exponential(double multiplier, int maxRetryCount) {
        return new RxBackoff(new Backoff.Builder()
                .setAlgorithm(new ExponentialAlgorithm(
                        ExponentialAlgorithm.DEFAULT_INTERVAL,
                        multiplier,
                        ExponentialAlgorithm.DEFAULT_MAX_INTERVAL,
                        ExponentialAlgorithm.DEFAULT_RANGE))
                .setMaxRetryCount(maxRetryCount)
                .build());
    }

    public static RxBackoff fixed(long interval, int maxRetryCount) {
        return new RxBackoff(new Backoff.Builder()
                .setAlgorithm(new FixedIntervalAlgorithm(interval, TimeUnit.MILLISECONDS))
                .setMaxRetryCount(maxRetryCount)
                .build());
    }

    public static RxBackoff random(long lowInterval, long highInterval, int maxRetryCount) {
        return new RxBackoff(new Backoff.Builder()
                .setAlgorithm(new RandomIntervalAlgorithm(
                        lowInterval,
                        highInterval,
                        1.0,
                        1.0,
                        highInterval))
                .setMaxRetryCount(maxRetryCount)
                .build());
    }

    public static RxBackoff of(BackoffAlgorithm algorithm, int maxRetryCount) {
        selectedAlgorithm = algorithm;
        mxRetryCount = maxRetryCount;
        return new RxBackoff(new Backoff.Builder()
                .setAlgorithm(algorithm)
                .setMaxRetryCount(maxRetryCount)
                .build());
    }

    public static BackoffAlgorithm getAlgorithm() {
        return selectedAlgorithm;
    }
    public static int getMaxRetry() {
        return mxRetryCount;
    }

    /**
     * Construct RxBackoff.
     *
     * @param backoff Backoff object with algorithm specified. You can build the back-off object
     *                using the {@link Backoff.Builder}.
     */
    public RxBackoff(Backoff backoff) {
        this(backoff, Schedulers.computation());
    }

    /**
     * Construct RxBackoff.
     *
     * @param backoff           Backoff object with algorithm specified. You can build the back-off object
     *                          using the {@link Backoff.Builder}.
     * @param intervalScheduler Scheduler used in backoff interval
     */
    public RxBackoff(Backoff backoff, Scheduler intervalScheduler) {
        this.backoff = backoff;
        this.intervalScheduler = intervalScheduler;
    }

    /**
     * Filters errors emitted by an ObservableSource by only emitting those that satisfy a specified
     * predicate.
     *
     * @param predicate a function that evaluates each error emitted by the source ObservableSource,
     *                  returning {@code true} if it passes the filter
     * @return RxBackoff instance
     */
    public RxBackoff filter(Predicate<Throwable> predicate) {
        this.filter = predicate;
        return this;
    }

    /**
     * Set callback function called every time before retry processing
     *
     * @param onRetry action to be executed at each retry
     * @return RxBackoff instance
     */
    public RxBackoff doOnRetry(BiConsumer<Throwable, Integer> onRetry) {
        this.onRetry = onRetry;
        return this;
    }

    /**
     * Set callback function called when abort retry
     *
     * @param onAbort action to be executed when abort
     * @return RxBackoff instance
     */
    public RxBackoff doOnAbort(Consumer<Throwable> onAbort) {
        this.onAbort = onAbort;
        return this;
    }

    public Function<Observable<Throwable>, ObservableSource<?>> observable() {
        return new Function<Observable<Throwable>, ObservableSource<?>>() {
            @Override
            public ObservableSource apply(Observable<Throwable> attempts) throws Exception {
                return attempts.flatMap(new Function<Throwable, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Throwable throwable) throws Throwable {
                        if (!filter.test(throwable)) {
                            return Observable.error(throwable);
                        }

                        long interval = backoff.interval();
                        if (interval != Backoff.ABORT) {
                            onRetry.accept(throwable, backoff.getRetryCount());
                            return Observable.timer(interval, TimeUnit.MILLISECONDS, intervalScheduler);
                        } else {
                            onAbort.accept(throwable);
                            return Observable.error(throwable);
                        }
                    }
                });
            }
        };
    }

    public Function<Flowable<Throwable>, Publisher<?>> flowable() {
        return new Function<Flowable<Throwable>, Publisher<?>>() {
            @Override
            public Publisher<?> apply(Flowable<Throwable> attempts) throws Exception {
                return attempts.flatMap(new Function<Throwable, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(Throwable throwable) throws Throwable {
                        if (!filter.test(throwable)) {
                            return Flowable.error(throwable);
                        }

                        long interval = backoff.interval();
                        if (interval != Backoff.ABORT) {
                            onRetry.accept(throwable, backoff.getRetryCount());
                            return Flowable.timer(interval, TimeUnit.MILLISECONDS, intervalScheduler);
                        } else {
                            onAbort.accept(throwable);
                            return Flowable.error(throwable);
                        }
                    }
                });
            }
        };
    }
}
